package com.atguigu.day01;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class Flink01_BatchWordCount {
    public static void main(String[] args) throws Exception {
        /**
         * spark:求wordCount
         * 1.创建spark环境
         * 2.获取文件中的数据 textFile
         * 3.将一行数据按照空格切分 切出每个单词 FlatMap
         * 4.将每个单词转为 Tuple2元组 kv键值对 k：单词 v：1  mapTo。。。。
         * 5.将相同单词的数据聚合到一块然后对value做累加 reduceByKey
         * 6.将结果打印到控制台 print
         * 7.调用启动算子 sc.start()
         * 8.关闭字段 stop()
         */

        //1.创建Flink的批执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        //2.获取文件中的数据
        DataSource<String> dataSource = env.readTextFile("input/word.txt");

        //3.将一行数据按照空格切分 切出每个单词
        FlatMapOperator<String, String> wordData = dataSource.flatMap(new MyFlatMap());

        //4.将每个单词转为 Tuple2元组 kv键值对 k：单词 v：1
        MapOperator<String, Tuple2<String, Integer>> wordToOneData = wordData.map(new MapFunction<String, Tuple2<String, Integer>>() {
            /**
             *
             *
             * @param value 经过上游（flatmap）处理过后的数据就是每一个单词
             * @return 将单词组成Tuple2元组返回
             *  因为Map方法是典型的UDF函数 一进一出 所以可以使用return返回
             * @throws Exception
             */
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
//                return new Tuple2<>(value,1);
                return Tuple2.of(value, 1);
            }
        });

        //5.将相同单词的数据聚合到一块
        UnsortedGrouping<Tuple2<String, Integer>> groupBy = wordToOneData.groupBy(0);
//        UnsortedGrouping<Tuple2<String, Integer>> groupBy = wordToOneData.groupBy("f0");
        // 6.然后对value做累加
        AggregateOperator<Tuple2<String, Integer>> result = groupBy.sum(1);

        //7.将结果打印到控制台
        result.print();
    }

    private static class MyFlatMap implements FlatMapFunction<String,String>{

        /**
         *  将一行数据按照空格切分 切出每个单词
         * @param value 输入的每一行数据
         * @param out 收集器，用来把数据发送至下游 通常是方法没有返回值的时候使用
         *            不使用return返回是因为 首先FlatMap相当于udtf函数一进多出，那么就意味着
         *            要返回n条数据，return只能返回一条数据 因此不适合用return方式返回，所以FlatMap这个
         *            方法没有返回值 是void 所以只能用Collector收集器将要返回的数据收集起来发送给下游
         * @throws Exception
         */
        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            //按照空格切分获取到每一个单词组成的字符串数组
            String[] words = value.split(" ");
            //遍历这个数组获取到每一个单词
            for (String word : words) {
                //将每一个单词发送至下游
                out.collect(word);
            }
        }
    }
}
